﻿/*
-----------------------------------------------------------------------------
This source file is part of Cell Cloud.

Copyright (c) 2009-2012 Cell Cloud Team - cellcloudproject@gmail.com

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in
all copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
THE SOFTWARE.
-----------------------------------------------------------------------------
*/

#include "CellStableHeaders.h"
#include "CellBlockingAcceptor.h"
#include "CellBlockingAcceptorThread.h"
#include "CellSession.h"
#include "CellMessage.h"
#include "CellStringConverter.h"
#include "CellLoggerManager.h"
#include "CellCachedThreadPool.h"
#include "CellUtil.h"

namespace cell
{

#define BA_MAXCONNUM 100
#define BA_BUFSIZE 8192

//-----------------------------------------------------------------------
BlockingAcceptor::BlockingAcceptor(void)
	: MessageAcceptor()
	, mMaxConnectNum(BA_MAXCONNUM)
	, mBufSize(BA_BUFSIZE)
	, mThread(NULL)
	, mExecutor(NULL)
	, mListener(INVALID_SOCKET)
{
	char poolName[64] = {0x0};
	sprintf(poolName, "BAThreadPool_%s", cell::peek(this).c_str());
	mExecutor = CELL_NEW CachedThreadPool(poolName, mMaxConnectNum / 4, mMaxConnectNum);
}
//-----------------------------------------------------------------------
BlockingAcceptor::~BlockingAcceptor(void)
{
	// 解除绑定
	unbind();

	forceClearupSessions();

	while (!mExpiredThreads.empty())
	{
		destroyRecvThreadDelay(NULL);
	}

	while (!mExpiredSessions.empty())
	{
		fireSessionDestroyed(NULL, NULL);
	}

	// 销毁所有过期会话
	{
		CELL_LOCK_MUTEX(mSessionMutex);

		if (!mExpiredSessions.empty())
		{
			ExpiredSession* es;
			ExpiredSessionList::iterator iter, end = mExpiredSessions.end();
			for (iter = mExpiredSessions.begin(); iter != end; ++iter)
			{
				es = (*iter);
				Session* session = es->session;
				SessionCapsule* capsule = es->capsule;

				CELL_DELETE session;
				CELL_DELETE capsule;
				CELL_DELETE es;
			}

			mExpiredSessions.clear();
		}
	}

	// 销毁所有过期线程
	{
		CELL_LOCK_MUTEX(mExpiredThreadMutex);

		if (!mExpiredThreads.empty())
		{
			RecvThreadList::iterator iter, end = mExpiredThreads.end();
			for (iter = mExpiredThreads.begin(); iter != end; ++iter)
			{
				CELL_DELETE (*iter);
			}

			mExpiredThreads.clear();
		}
	}

	if (NULL != mExecutor)
	{
		CELL_DELETE dynamic_cast<CachedThreadPool*>(mExecutor);
		mExecutor = NULL;
	}
}
//-----------------------------------------------------------------------
bool BlockingAcceptor::bind(uint port)
{
	Inet4Address address(port);
	return bind(address);
}
//-----------------------------------------------------------------------
bool BlockingAcceptor::bind(const InetAddress& addr)
{
	if (mListener != INVALID_SOCKET)
	{
		fireErrorOccurred(MessageHandler::EC_STATE_ERROR, NULL);
		return false;
	}

	if (!addr.isValid())
	{
		fireErrorOccurred(MessageHandler::EC_ADDRESS_INVALID, NULL);
		return false;
	}

	socket_t serv = INVALID_SOCKET;
	// 创建套接字
	if (INVALID_SOCKET == (serv = ::socket(AF_INET, SOCK_STREAM, 0)))
	{
		fireErrorOccurred(MessageHandler::EC_SOCK_FAILED, NULL);
		return false;
	}

	// 设置 socket 超时时间
	int timeout = 30000;
	if (::setsockopt(serv, SOL_SOCKET, SO_SNDTIMEO, (char *)&timeout, sizeof(timeout)) == SOCKET_ERROR)
	{
		char log[128] = {0x0};
		sprintf(log, "[CC] Sets socket sending timeout failed.");
		LoggerManager::getSingleton().log(log);
	}
	timeout = 30000;
	if (::setsockopt(serv, SOL_SOCKET, SO_RCVTIMEO, (char *)&timeout, sizeof(timeout)) == SOCKET_ERROR)
	{
		char log[128] = {0x0};
		sprintf(log, "[CC] Sets socket receiving timeout failed.");
		LoggerManager::getSingleton().log(log);
	}

	// 设置缓冲区
	/*int bufSize = 81960;
	if (::setsockopt(serv, SOL_SOCKET, SO_RCVBUF, (char *)&bufSize, sizeof(bufSize)) == SOCKET_ERROR)
	{
		LoggerManager::getSingleton().log("[CC] Sets socket revbuf failed.");
	}
	bufSize = 81960;
	if (::setsockopt(serv, SOL_SOCKET, SO_SNDBUF, (char *)&bufSize, sizeof(bufSize)) == SOCKET_ERROR)
	{
		LoggerManager::getSingleton().log("[CC] Sets socket sndbuf failed.");
	}*/

	// Keep alive
	int keepAlive = 1;
	if (::setsockopt(serv, SOL_SOCKET, SO_KEEPALIVE, (char *)&keepAlive, sizeof(keepAlive)) == SOCKET_ERROR)
	{
		LoggerManager::getSingleton().log("[CC] Sets socket option SO_KEEPALIVE failed.");
	}

	if (addr.isIPv4())
	{
		sockaddr_in sockAddr;

		memset(&sockAddr, 0, sizeof(sockAddr));
		sockAddr.sin_family = AF_INET;
		sockAddr.sin_port = htons(addr.getPort());
		if (addr.getHostAddress().empty())
			sockAddr.sin_addr.s_addr = htonl(INADDR_ANY);
		else
			sockAddr.sin_addr.s_addr = inet_addr(addr.getHostAddress().c_str());

		// 绑定地址和端口
		if (SOCKET_ERROR == ::bind(serv, (LPSOCKADDR)&sockAddr, sizeof(sockAddr)))
		{
			fireErrorOccurred(MessageHandler::EC_BIND_FAILED, NULL);
			closeSocket(serv);
			return false;
		}
	}
	else
	{
		sockaddr_in6 sockAddr6;

		memset(&sockAddr6, 0, sizeof(sockAddr6));
		sockAddr6.sin6_family = AF_INET6;
		sockAddr6.sin6_port = htons(addr.getPort());
		unsigned long destAddr;
		if (addr.getHostAddress().empty())
			destAddr = INADDR_ANY;
		else
			destAddr = inet_addr(addr.getHostAddress().c_str());
		memcpy(&sockAddr6.sin6_addr, &destAddr, sizeof(struct in6_addr));

		// 绑定地址和端口
		if (SOCKET_ERROR == ::bind(serv, (LPSOCKADDR)&sockAddr6, sizeof(sockAddr6)))
		{
			fireErrorOccurred(MessageHandler::EC_BIND_FAILED, NULL);
			closeSocket(serv);
			return false;
		}
	}

	// 启动监听
	if (SOCKET_ERROR == ::listen(serv, SOMAXCONN))
	{
		fireErrorOccurred(MessageHandler::EC_LISTEN_FAILED, NULL);
		closeSocket(serv);
		return false;
	}

	// 存储 Sock 句柄
	mListener = serv;
	mV4Address = addr;

	// 启动守护线程
	mThread = CELL_NEW BADaemonThread(this);
	mThread->start();

	return true;
}
//-----------------------------------------------------------------------
void BlockingAcceptor::unbind(void)
{
	if (mThread != NULL)
	{
		mThread->stop();
	}

	if (INVALID_SOCKET != mListener)
	{
		closeSocket(mListener);
		mListener = INVALID_SOCKET;
	}

	int count = 0;
	if (mThread != NULL)
	{
		while (mThread->isRunning())
		{
			Thread::sleep(10);
			++count;
			if (count > 3000)
			{
				break;
			}
		}

		CELL_DELETE mThread;
		mThread = NULL;
	}

	SessionList list;

	{
		// 关闭所有 Session

		CELL_LOCK_MUTEX(mSessionMutex);

		Sessions::iterator iter, end = mSessions.end();
		for (iter = mSessions.begin(); iter != end; ++iter)
		{
			list.push_back(iter->first);
		}
	}

	if (!list.empty())
	{
		SessionList::iterator iter, end = list.end();
		for (iter = list.begin(); iter != end; ++iter)
		{
			close((*iter));
		}
	}

	// 等待 Session 全部关闭
	count = 0;
	while (!mSessions.empty() || !mDesertedSessions.empty())
	{
		Thread::sleep(10);

		++count;
		if (count > 3000)
		{
			char log[256] = {0x0};
			sprintf(log, "[CC] Acceptor [%s] CAN NOT close some sessions. Session list length is %d , deserted session list length is %d .",
				mExecutor->getName().c_str(), mSessions.size(), mDesertedSessions.size());
			LoggerManager::getSingleton().log(log);

			break;
		}
	}

	forceClearupSessions();
}
//-----------------------------------------------------------------------
void BlockingAcceptor::setMaxConnectNum(int num)
{
	mMaxConnectNum = num;
	dynamic_cast<CachedThreadPool*>(mExecutor)->resetMaxPoolSize(mMaxConnectNum);
	dynamic_cast<CachedThreadPool*>(mExecutor)->resetCorePoolSize(mMaxConnectNum / 4);
}
//-----------------------------------------------------------------------
inline int BlockingAcceptor::getConnectNum(void)
{
	return dynamic_cast<CachedThreadPool*>(mExecutor)->getActiveCount();
}
//-----------------------------------------------------------------------
inline uint BlockingAcceptor::getPort(void) const
{
	return mV4Address.getPort();
}
//-----------------------------------------------------------------------
void BlockingAcceptor::write(Session* session, Message* message)
{
	uint32 dataSize = message->length();
	if (dataSize == 0)
	{
		// 通知错误
		fireErrorOccurred(MessageHandler::EC_WRITE_FAILED, session);
		return;
	}

	// 获取数据
	char* data = CELL_ALLOC_T(char, dataSize + 1, MEMCATEGORY_GENERAL);
	memset(data, 0x0, dataSize + 1);
	message->get(data);

	// 数据缓存
	char* buf = NULL;
	int size = 0;

	if (existDataMark())
	{
		size = dataSize + getHeadLength() + getTailLength();
		buf = CELL_ALLOC_T(char, size + 1, MEMCATEGORY_GENERAL);
		memset(buf, 0x0, size + 1);
		memcpy(buf, getHeadMark(), getHeadLength());
		memcpy(buf + getHeadLength(), data, dataSize);
		memcpy(buf + getHeadLength() + dataSize, getTailMark(), getTailLength());
	}
	else
	{
		size = dataSize;
		buf = CELL_ALLOC_T(char, size + 1, MEMCATEGORY_GENERAL);
		memset(buf, 0x0, size + 1);
		memcpy(buf, data, size);
	}

	CELL_FREE(data, MEMCATEGORY_GENERAL);

	SessionCapsule* cap = NULL;

	{
		CELL_LOCK_MUTEX(mSessionMutex);

		Sessions::iterator iter = mSessions.find(session);
		if (iter != mSessions.end())
		{
			cap = iter->second;
		}
	}

	if (NULL == cap)
	{
		// 释放 buf 内存
		CELL_FREE(buf, MEMCATEGORY_GENERAL);
		return;
	}

	bool error = false;

	{
		CELL_LOCK_MUTEX(cap->mutex);

		cap->buffer.push_back(buf);

		socket_t sock = cap->sock;

		while (false == cap->buffer.empty())
		{
			char* databuf = (*cap->buffer.begin());

			int ssize = 0;
			int ss = 0;
			while (ssize < size)
			{
				if (SOCKET_ERROR == (ss = ::send(sock, databuf + ssize, size - ssize, 0)))
				{
					error = true;
					break;
				}

				ssize += ss;
			}

			if (false == error)
			{
				databuf = (*cap->buffer.begin());
				cap->buffer.pop_front();
				CELL_FREE(databuf, MEMCATEGORY_GENERAL);
			}
			else
			{
				char log[128] = {0x0};
				sprintf(log, "[CC] Blocking acceptor write error @ %s:%d"
					, session->getAddress().getHostAddress().c_str()
					, session->getAddress().getPort());
				LoggerManager::getSingleton().log(log);
				break;
			}
		}
	}

	if (false == error)
	{
		fireMessageSent(session, message);
	}
	else
	{
		fireErrorOccurred(MessageHandler::EC_WRITE_FAILED, session);
	}

	//_ASSERTE(_CrtCheckMemory());
}
//-----------------------------------------------------------------------
void BlockingAcceptor::read(Message* message, Session* session)
{
	// Nothing
}
//-----------------------------------------------------------------------
void BlockingAcceptor::close(Session* session)
{
	SessionCapsule* capsule = NULL;

	CELL_LOCK_MUTEX(mSessionMutex);

	Sessions::iterator iter = mSessions.find(session);
	if (iter != mSessions.end())
	{
		capsule = iter->second;

		// 将 Session 加入被遗弃列表
		Sessions::iterator diter = mDesertedSessions.find(session);
		if (diter == mDesertedSessions.end())
		{
			mDesertedSessions.insert(Sessions::value_type(iter->first, iter->second));
		}

		// 从管理列表中删除
		mSessions.erase(iter);
	}

	if (NULL != capsule)
	{
		if (capsule->thread != NULL)
			capsule->thread->stop();

		closeSocket(capsule->sock);
	}
}
//-----------------------------------------------------------------------
void BlockingAcceptor::listSessionIds(SessionIdList& list)
{
	CELL_LOCK_MUTEX(mSessionMutex);

	Sessions::iterator iter, end = mSessions.end();
	for (iter = mSessions.begin(); iter != end; ++iter)
	{
		list.push_back(iter->first->getId());
	}
}
//-----------------------------------------------------------------------
Session* BlockingAcceptor::getSession(ulong id)
{
	CELL_LOCK_MUTEX(mSessionMutex);

	Sessions::const_iterator iter = mSessions.begin(), end = mSessions.end();
	while (iter != end)
	{
		if (iter->first->getId() == id)
		{
			return iter->first;
		}

		++iter;
	}

	return NULL;
}
//-----------------------------------------------------------------------
Session* BlockingAcceptor::fireSessionCreated(socket_t sock, const char* host, const uint port)
{
	Inet4Address address(host, port);
	Session* session = CELL_NEW Session(this, address);
	SessionCapsule* capsule = CELL_NEW SessionCapsule();
	capsule->sock = sock;

	try
	{
		{
			CELL_LOCK_MUTEX(mSessionMutex);

			mSessions.insert(Sessions::value_type(session, capsule));
		}

		if (mHandler != NULL)
		{
			mHandler->sessionCreated(session);
		}
	}
	catch (...)
	{
		CELL_DELETE session;
		session = NULL;

		CELL_DELETE capsule;
		capsule = NULL;
	}

	return session;
}
//-----------------------------------------------------------------------
void BlockingAcceptor::fireSessionDestroyed(Session* session, SessionCapsule* capsule)
{
	if (mHandler != NULL && session != NULL)
	{
		mHandler->sessionDestroyed(session);
	}

	uint64 time = Util::currentTimeMillis();

	CELL_LOCK_MUTEX(mSessionMutex);

	if (false == mExpiredSessions.empty())
	{
		ExpiredSession* es;
		ExpiredSessionList list;
		ExpiredSessionList::iterator iter, end = mExpiredSessions.end();
		for (iter = mExpiredSessions.begin(); iter != end; ++iter)
		{
			es = (*iter);
			if (time - es->time > 30000)
			{
				list.push_back(es);
			}
			else
			{
				break;
			}
		}

		if (false == list.empty())
		{
			ExpiredSessionList::iterator find;
			end = list.end();
			for (iter = list.begin(); iter != end; ++iter)
			{
				es = (*iter);
				find = std::find(mExpiredSessions.begin(), mExpiredSessions.end(), es);
				mExpiredSessions.erase(find);

				Session* s = es->session;
				SessionCapsule* c = es->capsule;
				CELL_DELETE s;
				CELL_DELETE c;
				CELL_DELETE es;
			}
		}
	}

	if (NULL == session)
		return;

	bool find = false;
	ExpiredSessionList::iterator iter, end = mExpiredSessions.end();
	for (iter = mExpiredSessions.begin(); iter != end; ++iter)
	{
		if ((*iter)->session == session)
		{
			find = true;
			break;
		}
	}

	if (false == find)
	{
		ExpiredSession* es = CELL_NEW ExpiredSession();
		es->time = time;
		es->session = session;
		es->capsule = capsule;
		mExpiredSessions.push_back(es);
	}
}
//-----------------------------------------------------------------------
void BlockingAcceptor::fireSessionOpened(Session* session)
{
	SessionCapsule* capsule = NULL;

	{
		CELL_LOCK_MUTEX(mSessionMutex);

		Sessions::const_iterator iter = mSessions.find(session);
		if (iter != mSessions.end())
		{
			capsule = iter->second;
		}
	}

	if (capsule != NULL)
	{
		if (mHandler != NULL)
		{
			mHandler->sessionOpened(session);
		}
	}
}
//-----------------------------------------------------------------------
void BlockingAcceptor::fireSessionClosed(Session* session)
{
	SessionCapsule* capsule = NULL;
	bool notifyEvent = false;

	{
		CELL_LOCK_MUTEX(mSessionMutex);

		Sessions::iterator iter = mDesertedSessions.find(session);
		if (iter != mDesertedSessions.end())
		{
			capsule = iter->second;

			mDesertedSessions.erase(iter);
		}

		if (capsule != NULL)
		{
			closeSocket(capsule->sock);
			notifyEvent = true;
		}
		else
		{
			Sessions::iterator iter = mSessions.find(session);
			if (iter != mSessions.end())
			{
				capsule = iter->second;

				mSessions.erase(iter);
			}

			if (capsule != NULL)
			{
				// 停止线程
				if (capsule->thread != NULL)
					capsule->thread->stop();

				closeSocket(capsule->sock);

				notifyEvent = true;
			}
		}
	}

	if (notifyEvent)
	{
		if (mHandler != NULL)
		{
			mHandler->sessionClosed(session);
		}
	}

	if (capsule != NULL)
	{
		// 不能直接在此删除线程对象，延迟销毁
		if (capsule->thread != NULL)
		{
			destroyRecvThreadDelay(capsule->thread);
		}

		fireSessionDestroyed(session, capsule);
	}

#if CELL_DEBUG_MODE
	char msg[128] = {0x0};
	sprintf(msg, "[CC] Acceptor (%d) connection number : %d [Closed]", mV4Address.getPort(), getConnectNum());
	LoggerManager::getSingleton().log(msg);
#endif // CELL_DEBUG_MODE
}
//-----------------------------------------------------------------------
void BlockingAcceptor::fireMessageReceived(Session* session, Message* message)
{
	SessionCapsule* capsule = NULL;

	{
		CELL_LOCK_MUTEX(mSessionMutex);

		Sessions::const_iterator iter = mSessions.find(session);
		if (iter != mSessions.end())
		{
			capsule = iter->second;
		}
	}

	if (capsule != NULL)
	{
		if (mHandler != NULL)
		{
			mHandler->messageReceived(session, message);
		}
	}
}
//-----------------------------------------------------------------------
void BlockingAcceptor::fireMessageSent(Session* session, Message* message)
{
	SessionCapsule* capsule = NULL;

	{
		CELL_LOCK_MUTEX(mSessionMutex);

		Sessions::const_iterator iter = mSessions.find(session);
		if (iter != mSessions.end())
		{
			capsule = iter->second;
		}
	}

	if (capsule != NULL)
	{
		if (mHandler != NULL)
		{
			mHandler->messageSent(session, message);
		}
	}
}
//-----------------------------------------------------------------------
void BlockingAcceptor::fireErrorOccurred(uint16 errorCode, Session* session)
{
	SessionCapsule* capsule = NULL;

	{
		CELL_LOCK_MUTEX(mSessionMutex);

		Sessions::const_iterator iter = mSessions.find(session);
		if (iter != mSessions.end())
		{
			capsule = iter->second;
		}
	}

	if (capsule != NULL)
	{
		if (mHandler != NULL)
		{
			mHandler->errorOccurred(errorCode, session);
		}
	}
}
//-----------------------------------------------------------------------
bool BlockingAcceptor::startRecvThread(Session* session)
{
	BARecvThread* thread = NULL;
	SessionCapsule* capsule = NULL;

	{
		CELL_LOCK_MUTEX(mSessionMutex);

	#if CELL_DEBUG_MODE
		char log[128] = {0x0};
		sprintf(log, "[CC] Blocking acceptor (%d) session number: %d", mV4Address.getPort(), mSessions.size());
		LoggerManager::getSingleton().log(log);
	#endif // CELL_DEBUG_MODE

		Sessions::iterator iter = mSessions.find(session);
		if (iter != mSessions.end())
		{
			bool error = false;
			try
			{
				thread = CELL_NEW BARecvThread(this, session);
			}
			catch (...)
			{
				error = true;
			}

			if (false == error)
			{
				SessionCapsule* capsule = iter->second;
				capsule->thread = thread;
			}
			else
			{
				thread = NULL;

				capsule = iter->second;

				mSessions.erase(iter);
			}
		}
	}

	if (thread != NULL)
	{
		// 执行任务
		if (mExecutor->execute(thread))
		{
			return true;
		}
		else
		{
			SessionCapsule* capsule = NULL;

			{
				CELL_LOCK_MUTEX(mSessionMutex);

				Sessions::iterator iter = mSessions.find(session);
				if (iter != mSessions.end())
				{
					thread->stop();

					capsule = iter->second;

					mSessions.erase(iter);
				}

				destroyRecvThreadDelay(thread);

				if (NULL != capsule)
					fireSessionDestroyed(session, capsule);
			}

			return false;
		}
	}
	else
	{
		if (NULL != capsule)
			fireSessionDestroyed(session, capsule);

		return false;
	}
}
//-----------------------------------------------------------------------
void BlockingAcceptor::destroyRecvThreadDelay(BARecvThread* thread)
{
	uint64 time = Util::currentTimeMillis();

	CELL_LOCK_MUTEX(mExpiredThreadMutex);

	// 销毁非运行状态的线程
	if (false == mExpiredThreads.empty())
	{
		RecvThreadList list;
		RecvThreadList::iterator iter, end = mExpiredThreads.end();
		for (iter = mExpiredThreads.begin(); iter != end; ++iter)
		{
			BARecvThread* t = (*iter);
			if (false == t->isRunning() && time - t->expiredTime >= 3000)
				list.push_back(t);
		}

		RecvThreadList::iterator titer;
		end = list.end();
		for (iter = list.begin(); iter != end; ++iter)
		{
			BARecvThread* t = (*iter);
			titer = std::find(mExpiredThreads.begin(), mExpiredThreads.end(), t);
			mExpiredThreads.erase(titer);
			CELL_DELETE t;
		}
	}

	if (NULL == thread)
		return;

	mExecutor->remove(thread);

	RecvThreadList::iterator iter = std::find(mExpiredThreads.begin(), mExpiredThreads.end(), thread);
	if (iter == mExpiredThreads.end())
	{
		thread->expiredTime = time;

		mExpiredThreads.push_back(thread);
	}
}
//-----------------------------------------------------------------------
socket_t BlockingAcceptor::getSock(Session* session)
{
	CELL_LOCK_MUTEX(mSessionMutex);

	Sessions::iterator iter = mSessions.find(session);
	if (iter != mSessions.end())
	{
		SessionCapsule* capsule = iter->second;
		return capsule->sock;
	}

	return INVALID_SOCKET;
}
//-----------------------------------------------------------------------
void BlockingAcceptor::forceClearupSessions(void)
{
	CELL_LOCK_MUTEX(mSessionMutex);

	Session* session = NULL;
	SessionCapsule* capsule = NULL;
	Sessions::iterator iter, end;

	end = mSessions.end();
	for (iter = mSessions.begin(); iter != end; ++iter)
	{
		session = iter->first;
		capsule = iter->second;

		if (capsule->thread != NULL)
		{
			capsule->thread->stop();
			closeSocket(capsule->sock);

			destroyRecvThreadDelay(capsule->thread);
			capsule->thread = NULL;
		}

		fireSessionDestroyed(session, capsule);
	}
	mSessions.clear();

	end = mDesertedSessions.end();
	for (iter = mDesertedSessions.begin(); iter != end; ++iter)
	{
		session = iter->first;
		capsule = iter->second;

		if (capsule->thread != NULL)
		{
			destroyRecvThreadDelay(capsule->thread);
		}

		fireSessionDestroyed(session, capsule);
	}
	mDesertedSessions.clear();
}


//-----------------------------------------------------------------------
BlockingAcceptor::SessionCapsule::SessionCapsule(void)
	: thread(NULL)
	, sock(INVALID_SOCKET)
{
}
//-----------------------------------------------------------------------
BlockingAcceptor::SessionCapsule::~SessionCapsule(void)
{
	CELL_LOCK_MUTEX(mutex);

	if (false == buffer.empty())
	{
		std::list<char*>::iterator iter, end = buffer.end();
		for (iter = buffer.begin(); iter != end; ++iter)
		{
			char* buf = (*iter);
			CELL_FREE(buf, MEMCATEGORY_GENERAL);
		}

		buffer.clear();
	}
}


//-----------------------------------------------------------------------
BlockingAcceptor::ExpiredSession::ExpiredSession(void)
	: session(NULL)
	, capsule(NULL)
	, time(0)
{
}
//-----------------------------------------------------------------------
BlockingAcceptor::ExpiredSession::~ExpiredSession(void)
{
}

} // end namespace cell
